package io.github.fb513.transport.queue;

import com.fasterxml.jackson.databind.JsonNode;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Delivery;
import io.github.fb513.common.model.BaseData;
import io.github.fb513.common.model.MessageInfo;
import io.github.fb513.common.util.ImUtils;
import io.github.fb513.transport.config.BootstrapConfig;
import io.github.fb513.transport.config.ImConfig;
import io.github.fb513.transport.exception.InitServiceException;
import io.github.fb513.transport.model.UserSession;
import io.github.fb513.transport.util.JsonUtils;
import io.github.fb513.transport.util.SessionHolder;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author fengbo
 * @since 2023/12/25 15:28
 */
@Slf4j
public class MessageQueueReceiver {

    private final String exchange;
    private final BootstrapConfig bootstrapConfig;
    private final ConnectionFactory connectionFactory;

    public MessageQueueReceiver(String exchange, BootstrapConfig bootstrapConfig, ConnectionFactory connectionFactory) {
        this.exchange = exchange;
        this.bootstrapConfig = bootstrapConfig;
        this.connectionFactory = connectionFactory;
    }

    public void listen() {
        try {
            ImConfig imConfig = bootstrapConfig.getIm();
            String brokerKey = ImUtils.getBrokerKey(imConfig.getIpNum(), imConfig.getPort());
            Channel channel = connectionFactory.newConnection().createChannel();
            channel.exchangeDeclare(exchange, "direct", true, false, null);
            channel.queueDeclare(brokerKey, true, false, false, null);
            channel.queueBind(brokerKey, exchange, brokerKey);
            channel.basicConsume(brokerKey, false, (consumerTag, message) -> {
                try {
                    parseMessage(consumerTag, message);
                    channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
                } catch (Exception e) {
                    log.error("消息处理异常", e);
                    channel.basicNack(message.getEnvelope().getDeliveryTag(),false,false);
                }
            }, consumerTag -> {});
        } catch (IOException | TimeoutException e) {
            throw new InitServiceException(e);
        }
    }

    private void parseMessage(String consumerTag, Delivery message) {
        String msg = new String(message.getBody());
        JsonNode jsonNode = JsonUtils.parseJsonToJsonNode(msg);
        MessageInfo.Header header = JsonUtils.convertToPojo(jsonNode.get("header"), MessageInfo.Header.class);
        int command = header.getCommand();
        BaseData<?> baseData = JsonUtils.convertToPojo(jsonNode.get("content"), BaseData.class);
        UserSession session = SessionHolder.getSession(baseData.getUserId(), baseData.getClientType());
        session.sendMsg(MessageInfo.getJsonMessage(command, baseData.getContent()));
    }
}
